/**
 * Copyright (C) 2010-2016 Alibaba Group Holding Limited
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package consumer;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;

import config.MqConfig;
import net.sf.json.JSONObject;
import net.sf.json.JSONArray;
import org.apache.thrift.*;
import org.apache.thrift.protocol.*;
import org.apache.thrift.transport.*;
import Services.RocketMqConsumer.*;
import org.apache.log4j.Logger;
import org.ini4j.Ini;
import org.ini4j.Profile.Section;
import java.io.FileReader;
import java.io.IOException;

/**
 * MQ消息处理类
 */
public class MessageListenerImpl implements MessageListener {

    private static Logger logger = Logger.getLogger(MessageListenerImpl.class);
    private static JSONArray conf_json = null;

    private  JSONObject getPHPServe(){
        if(conf_json == null){
            Ini ini = new Ini();
            try {
                //加载配置文件
                ini.load(new FileReader(MqConfig.config_file));
                Section consumer_section = ini.get(MqConfig.consumer_section);
                conf_json = JSONArray.fromObject(consumer_section.get("PHP_SERVE"));
            }catch (IOException e){
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
        int index = (int) (Math.random() * conf_json.size());
        JSONObject obj = conf_json.getJSONObject(index);
        return obj;
    }

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        Action action = Action.ReconsumeLater;
        TTransport transport = null;
        try {
            logger.info(message);
            System.out.println(message.getMsgID());
            JSONObject obj = this.getPHPServe();
            String host = obj.get("host").toString();
            Integer port = obj.getInt("port");
            transport = new TSocket(host, port, 30000);
            // 协议要和服务端一致
            TProtocol protocol = new TBinaryProtocol(transport);
            RocketMqConsumer.Client client = new RocketMqConsumer.Client(protocol);
            transport.open();
            MessageInfo messageInfo = new MessageInfo();
            messageInfo.message_id = message.getMsgID();
            messageInfo.topic = message.getTopic();
            messageInfo.tags = message.getTag();
            messageInfo.msg = new String(message.getBody());
            messageInfo.key = message.getKey();
            Result result = client.receiveMsg(messageInfo);
            if (result.code == 0){
                action = Action.CommitMessage;
            }
        } catch (TTransportException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        } catch (TException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        } finally {
            if (null != transport) {
                transport.close();
            }
        }
        return action;
    }
}
